package com.desheng.bigdata.flink.stream.transformation

import com.desheng.bigdata.flink.stream.source.Person
import org.apache.flink.streaming.api.scala._

object _06Transformation4ReduceOps {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        //按照province对人群进行分组，
        val persons = env.fromCollection(List(
            Person("张三", 18, 0, "河南"),
            Person("李四", 19, 1, "山东"),
            Person("王五", 20, 0, "河南"),
            Person("周七", 21, 0, "山西"),
            Person("赵六", 20, 1, "山东")
        ))

        persons.map(person => (person.province, 1))
                    .keyBy(0)
                    .reduce((pv1, pv2) => {
                        println(s"pv1: ${pv1}-----pv2: ${pv2}")
                        (pv1._1, pv1._2 + pv2._2)
                    }).print()

        env.execute(s"${_06Transformation4ReduceOps.getClass.getSimpleName}")
    }
}
